package top.guoziyang.mydb.backend.im;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import top.guoziyang.mydb.backend.common.SubArray;
import top.guoziyang.mydb.backend.dm.dataItem.DataItem;
import top.guoziyang.mydb.backend.tm.TransactionManagerImpl;
import top.guoziyang.mydb.backend.utils.Parser;

/**
 * Node结构如下：
 * [LeafFlag (1 Byte)][KeyNumber (2 Bytes)][SiblingUid (8 Bytes)]
 * [Son0 (8 Bytes)][Key0 (8 Bytes)][Son1 (8 Bytes)][Key1 (8 Bytes)]...[SonN (8 Bytes)][KeyN (8 Bytes)]
 */
public class Node {
    static final int IS_LEAF_OFFSET = 0; //叶子节点标志位,0
    static final int NO_KEYS_OFFSET = IS_LEAF_OFFSET+1; //存储当前节点中键（Key）的数量,1
    static final int SIBLING_OFFSET = NO_KEYS_OFFSET+2; // 存储兄弟节点在 DM 中的 UID，3
    static final int NODE_HEADER_SIZE = SIBLING_OFFSET+8; // 节点头部的大小: 3+8=11

    static final int BALANCE_NUMBER = 32; // 节点的平衡因子,即节点中允许存储的最多键值对数量的一半

    //一个节点的总大小，包含节点头部、子节点UID和键值
    static final int NODE_SIZE = NODE_HEADER_SIZE + (2*8)*(BALANCE_NUMBER*2+2);
    /*
    /   BALANCE_NUMBER*2表示一个节点最多可以存储的键值对数量,在 B+树中，节点有N个键时，需要 N+1 个子节点指针,
        因此，当节点中最多存储BALANCE_NUMBER*2个键时，需要存储 BALANCE_NUMBER*2+1 个子节点指针
        再加 1 是为了额外预留空间（如分裂操作时可能需要一个额外的子节点指针）。
        在节点中，每个键值对由以下两部分组成：子节点指针（Son）：指向子节点的地址，占用 8 字节 ，键值（Key）：存储键值，用于查找匹配，占用 8 字节，所以没对键值对的大小为16字节
     */

    BPlusTree tree;
    DataItem dataItem;
    SubArray raw;
    long uid;

    /**
     * 设置节点的叶子节点标志
     */
    static void setRawIsLeaf(SubArray raw, boolean isLeaf) {
        if(isLeaf) {
            raw.raw[raw.start + IS_LEAF_OFFSET] = (byte)1;
        } else {
            raw.raw[raw.start + IS_LEAF_OFFSET] = (byte)0;
        }
    }

    /**
     * 判断节点是否是叶子节点,是返回true
     */
    static boolean getRawIfLeaf(SubArray raw) {
        return raw.raw[raw.start + IS_LEAF_OFFSET] == (byte)1;
    }

    /**
     * 设置节点中的键数量
     */
    static void setRawNoKeys(SubArray raw, int noKeys) {
        System.arraycopy(Parser.short2Byte((short)noKeys), 0, raw.raw, raw.start+NO_KEYS_OFFSET, 2);
    }

    /**
     * 获取节点中的键数量
     */
    static int getRawNoKeys(SubArray raw) {
        return (int)Parser.parseShort(Arrays.copyOfRange(raw.raw, raw.start+NO_KEYS_OFFSET, raw.start+NO_KEYS_OFFSET+2));
    }

    /**
     * 设置节点的兄弟节点的uid
     */
    static void setRawSibling(SubArray raw, long sibling) {
        System.arraycopy(Parser.long2Byte(sibling), 0, raw.raw, raw.start+SIBLING_OFFSET, 8);
    }

    /**
     * 获取节点的兄弟节点的uid
     */
    static long getRawSibling(SubArray raw) {
        return Parser.parseLong(Arrays.copyOfRange(raw.raw, raw.start+SIBLING_OFFSET, raw.start+SIBLING_OFFSET+8));
    }

    /**
     * 设置第 kth 个子节点的UID
     */
    static void setRawKthSon(SubArray raw, long uid, int kth) {
        int offset = raw.start+NODE_HEADER_SIZE+kth*(8*2);
        System.arraycopy(Parser.long2Byte(uid), 0, raw.raw, offset, 8);
    }

    /**
     * 获取第 kth 个子节点的UID
     */
    static long getRawKthSon(SubArray raw, int kth) {
        int offset = raw.start+NODE_HEADER_SIZE+kth*(8*2);
        return Parser.parseLong(Arrays.copyOfRange(raw.raw, offset, offset+8));
    }

    /**
     * 设置第 kth 个子节点的键值Key
     */
    static void setRawKthKey(SubArray raw, long key, int kth) {
        int offset = raw.start+NODE_HEADER_SIZE+kth*(8*2)+8;
        System.arraycopy(Parser.long2Byte(key), 0, raw.raw, offset, 8);
    }

    /**
     * 获取第 kth 个子节点的键值Key
     */
    static long getRawKthKey(SubArray raw, int kth) {
        int offset = raw.start+NODE_HEADER_SIZE+kth*(8*2)+8;
        return Parser.parseLong(Arrays.copyOfRange(raw.raw, offset, offset+8));
    }

    /**
     * 将源节点（from）从第 kth 个键值对开始的数据复制到目标节点（to）
     */
    static void copyRawFromKth(SubArray from, SubArray to, int kth) {
        int offset = from.start+NODE_HEADER_SIZE+kth*(8*2);
        System.arraycopy(from.raw, offset, to.raw, to.start+NODE_HEADER_SIZE, from.end-offset);
    }

    /**
     * 在同一个节点内，从第 kth+1 个键值对开始，将所有后续数据向后移动，为插入新键值对预留空间
     */
    static void shiftRawKth(SubArray raw, int kth) {
        int begin = raw.start+NODE_HEADER_SIZE+(kth+1)*(8*2);
        int end = raw.start+NODE_SIZE-1;
        for(int i = end; i >= begin; i --) {
            raw.raw[i] = raw.raw[i-(8*2)];
        }
    }

    /**
     * 创建一个新的非叶子根节点的原始字节数组（raw），并初始化其头部信息、子节点指针和键值
     */
    static byte[] newRootRaw(long left, long right, long key)  {
        SubArray raw = new SubArray(new byte[NODE_SIZE], 0, NODE_SIZE);

        setRawIsLeaf(raw, false); // 设置为非叶子节点
        setRawNoKeys(raw, 2); // 初始化键数量为 2
        setRawSibling(raw, 0); // 没有兄弟节点，兄弟指针为 0
        setRawKthSon(raw, left, 0); // 设置左子节点UID
        setRawKthKey(raw, key, 0); // 设置第一个键值
        setRawKthSon(raw, right, 1); // 设置右子节点UID
        setRawKthKey(raw, Long.MAX_VALUE, 1); // 设置最大键值（哨兵值）

        return raw.raw;
    }

    /**
     * 创建一个新的空根节点（叶子节点），用于表示树为空的初始状态
     * 该方法用于表示 B+树为空时的初始根节点状态
     */
    static byte[] newNilRootRaw()  {
        SubArray raw = new SubArray(new byte[NODE_SIZE], 0, NODE_SIZE);

        setRawIsLeaf(raw, true); // 设置为叶子节点
        setRawNoKeys(raw, 0);  // 键值数量初始化为 0
        setRawSibling(raw, 0); // 没有兄弟节点，兄弟指针为 0

        return raw.raw;
    }

    /**
     * 从存储中加载指定的节点（通过 uid ），并将其包装成一个 Node 对象
     */
    static Node loadNode(BPlusTree bTree, long uid) throws Exception {
        DataItem di = bTree.dm.read(uid);
        assert di != null;
        Node n = new Node();
        n.tree = bTree;
        n.dataItem = di;
        n.raw = di.data();
        n.uid = uid;
        return n;
    }

    /**
     * 释放资源
     */
    public void release() {
        dataItem.release();
    }

    /**
     * 判断是否是叶子节点
     */
    public boolean isLeaf() {
        dataItem.rLock();
        try {
            return getRawIfLeaf(raw);
        } finally {
            dataItem.rUnLock();
        }
    }

    /**
     * 在非叶子节点中搜索目标键值，返回结果包含以下两种情况：
     * 1.如果目标键值落在当前节点的某个范围内，返回对应子节点的 uid。
     * 2.如果目标键值超出了当前节点的范围，返回当前节点的兄弟节点 siblingUid。
     */
    class SearchNextRes {
        long uid; // 子节点UID
        long siblingUid; // 当前节点的兄弟节点 siblingUid
    }

    public SearchNextRes searchNext(long key) {
        dataItem.rLock();
        try {
            SearchNextRes res = new SearchNextRes();
            int noKeys = getRawNoKeys(raw);
            // 遍历节点中的键值
            for(int i = 0; i < noKeys; i ++) {
                long ik = getRawKthKey(raw, i);
                // 如果目标键值小于当前键值，则说明目标键值应当落在第 i 个键值对应的子节点中
                if(key < ik) {
                    // 获取对应子节点的UID,不需要转移到兄弟节点
                    res.uid = getRawKthSon(raw, i);
                    res.siblingUid = 0;
                    return res;
                }
            }
            //如果遍历完所有键值未找到匹配范围，说明目标键值大于当前节点的所有键值
            // 设置 res.siblingUid 为当前节点的兄弟节点的UID，用于继续搜索.
            res.uid = 0;
            res.siblingUid = getRawSibling(raw);
            return res;

        } finally {
            dataItem.rUnLock();
        }
    }

    /**
     *  B+树叶子节点的范围查询逻辑
     */
    class LeafSearchRangeRes {
        List<Long> uids;
        long siblingUid;
    }

    /**
     * 根据给定的键值范围 [leftKey, rightKey]，在当前叶子节点中查找符合条件的子节点（uid），并返回结果
     */
    public LeafSearchRangeRes leafSearchRange(long leftKey, long rightKey) {
        dataItem.rLock();
        try {
            int noKeys = getRawNoKeys(raw);
            int kth = 0;
            // 循环读取子节点的key，直到找到第一个key>= leftKey的子节点
            while(kth < noKeys) {
                long ik = getRawKthKey(raw, kth);
                if(ik >= leftKey) {
                    break;
                }
                kth ++;
            }
            List<Long> uids = new ArrayList<>();
            // 继续读取子节点，把所有key <= rightKey的子节点的UID添加到uids里
            while(kth < noKeys) {
                long ik = getRawKthKey(raw, kth);
                if(ik <= rightKey) {
                    uids.add(getRawKthSon(raw, kth));
                    kth ++;
                } else {
                    break;
                }
            }

            long siblingUid = 0;
            // 如果键值范围 [leftKey, rightKey]部分落在当前节点内，此时kth >= noKeys，需要转移到兄弟节点中继续查询
            if(kth == noKeys) {
                siblingUid = getRawSibling(raw);
            }
            LeafSearchRangeRes res = new LeafSearchRangeRes();
            res.uids = uids;
            res.siblingUid = siblingUid;
            return res;
        } finally {
            dataItem.rUnLock();
        }
    }

    /**
     *  B+树入操作的核心逻辑，包括在当前节点插入键值对、判断是否需要分裂节点以及分裂节点后的结果处理
     */
    class InsertAndSplitRes {
        long siblingUid ;// 当前节点的兄弟节点 UID（如果插入失败需要跳转兄弟节点）
        long  newSon;  // 分裂后新节点的 UID
        long  newKey; // 分裂后提升到父节点的新键值
    }

    /**
     *  在当前节点插入键值对（uid, key），并在必要时进行节点分裂，返回插入及分裂的结果。
     */
    public InsertAndSplitRes insertAndSplit(long uid, long key) throws Exception {
        boolean success = false;
        Exception err = null;
        InsertAndSplitRes res = new InsertAndSplitRes();

        dataItem.before(); // 插入操作前进行备份
        try {
            success = insert(uid, key);
            // 如果插入失败（当前节点已满，且右兄弟节点存在），返回兄弟节点 UID
            if(!success) {
                res.siblingUid = getRawSibling(raw);
                return res;
            }
            // 如果插入成功，且需要分裂
            if(needSplit()) {
                try {
                    SplitRes r = split(); // 分裂节点
                    res.newSon = r.newSon;  // 设置新节点 UID
                    res.newKey = r.newKey; // 设置提升到父节点的新键值
                    return res;
                } catch(Exception e) {
                    err = e;
                    throw e;
                }
            } else {
                // 插入成功，且不需要分裂
                return res;
            }
        } finally {
            // 如果插入成功，且没有异常，提交事务
            if(err == null && success) {
                dataItem.after(TransactionManagerImpl.SUPER_XID);
            } else {  // 否则，回滚事务
                dataItem.unBefore();
            }
        }
    }

    /**
     *  在当前节点插入键值对（uid, key），返回插入是否成功的布尔值。
     */
    private boolean insert(long uid, long key) {
        int noKeys = getRawNoKeys(raw);
        int kth = 0;
        while(kth < noKeys) {
            long ik = getRawKthKey(raw, kth);
            if(ik < key) {
                kth ++;
            } else {
                break;
            }
        }
        // 如果当前节点已满，且存在兄弟节点，插入失败
        if(kth == noKeys && getRawSibling(raw) != 0) return false;

        if(getRawIfLeaf(raw)) { // 如果是叶子节点
            shiftRawKth(raw, kth);  // 移动数据，为插入腾出空间
            setRawKthKey(raw, key, kth); // 插入新键值
            setRawKthSon(raw, uid, kth); // 插入新子节点UID
            setRawNoKeys(raw, noKeys+1); // 更新键值数量
        } else {  // 如果是非叶子节点
            long kk = getRawKthKey(raw, kth); // 记录原位置键值
            setRawKthKey(raw, key, kth);  // 替换原位置键值
            shiftRawKth(raw, kth+1);  // 移动后续数据
            setRawKthKey(raw, kk, kth+1); // 将原键值插入到后续位置
            setRawKthSon(raw, uid, kth+1); // 插入新子节点UID
            setRawNoKeys(raw, noKeys+1); // 更新键值数量
        }
        return true;
    }


    /**
     * 判断是否需要分裂
     */
    private boolean needSplit() {
        return BALANCE_NUMBER*2 == getRawNoKeys(raw);
    }

    /**
     * B+树节点分裂逻辑
     */
    class SplitRes {
        long newSon, newKey;
    }

    /**
     * 当节点的键值数量达到上限（BALANCE_NUMBER * 2）时，将节点分裂为两个节点，保持 B+树的平衡性。
     */
    private SplitRes split() throws Exception {
        // 创建一个新的 SubArray 对象 nodeRaw,用于存储分裂出的新节点的数据
        SubArray nodeRaw = new SubArray(new byte[NODE_SIZE], 0, NODE_SIZE);
        setRawIsLeaf(nodeRaw, getRawIfLeaf(raw)); // 新节点的叶子标志与原节点一致。
        setRawNoKeys(nodeRaw, BALANCE_NUMBER); // 新节点的初始键值数量为 BALANCE_NUMBER。
        setRawSibling(nodeRaw, getRawSibling(raw)); //新节点的兄弟节点UID继承原节点的兄弟节点
        copyRawFromKth(raw, nodeRaw, BALANCE_NUMBER); //从原节点的第 BALANCE_NUMBER 个键值开始，将其数据拷贝到新节点
        long son = tree.dm.insert(TransactionManagerImpl.SUPER_XID, nodeRaw.raw); //将新节点的数据通过dm写入缓存中并返回新节点的UID
        setRawNoKeys(raw, BALANCE_NUMBER); // 更新原节点的键值数量为 BALANCE_NUMBER（保留前一半键值）
        setRawSibling(raw, son); // 更新原节点的兄弟指针，使其指向新节点
        // 创建 SplitRes 对象存储分裂结果
        SplitRes res = new SplitRes();
        res.newSon = son; //新节点的 UID
        res.newKey = getRawKthKey(nodeRaw, 0); //新节点的第一个键值，它是其管理范围的最小值，用于提升到父节点，父节点将它作为新的索引键值。
        return res;
    }


    /**
     * 将节点的内容转换为字符串，用于调试和打印
     */
    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("Is leaf: ").append(getRawIfLeaf(raw)).append("\n");
        int KeyNumber = getRawNoKeys(raw);
        sb.append("KeyNumber: ").append(KeyNumber).append("\n");
        sb.append("sibling: ").append(getRawSibling(raw)).append("\n");
        for(int i = 0; i < KeyNumber; i ++) {
            sb.append("son: ").append(getRawKthSon(raw, i)).append(", key: ").append(getRawKthKey(raw, i)).append("\n");
        }
        return sb.toString();
    }

}
